package com.xinqing.bigdata.test.datastream

import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.util.Collector

/**
  * @Author:CHQ
  * @Date:2020 /6/22 15:32
  * @Description
  */
class TempChangeAlterFunction() extends ProcessFunction[SensorReading, (String, Double, Double)] {

  //存储上一次温度的状态值
  lazy val lastTemp: ValueState[Double] = getRuntimeContext.getState(new ValueStateDescriptor[Double]("lastTemp", classOf[Double]))

  override def processElement(value: SensorReading, ctx: ProcessFunction[SensorReading, (String, Double, Double)]#Context, out: Collector[(String, Double, Double)]): Unit = {

    val curTemp = value.temperature

    if ((curTemp - lastTemp.value()).abs > 10) {
      out.collect(value.id, lastTemp.value(), curTemp)
      lastTemp.update(curTemp)
    }

  }
}
